#include "Subscriber_profile.h"
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include "Log/Logger.h"
using namespace eprosima::fastdds::dds;
using namespace eprosima::fastdds::rtps;
extern time_t getTimestamp();
Subscriber_profile::Subscriber_profile() : participant_(nullptr), type_(new MessagePubSubType())
{
}

Subscriber_profile::~Subscriber_profile()
{
	auto it = subscribers_.begin();
	while (it != subscribers_.end())
	{
		auto listener = it->second->get_listener();
		delete listener;
		++it;
	}
	if (participant_)
	{
		participant_->delete_contained_entities();
		DomainParticipantFactory::get_instance()->delete_participant(participant_);
	}
}

bool Subscriber_profile::init(std::string xml_path, std::string participant_profile,
							  std::string topic_profile, std::string datareader_profile, std::string topic_name, std::function<void(Message)>callbk)
{
	if (subscribers_.count(topic_name))
	{
		return true;
	}
	if (!participant_)
	{
		auto ret = DomainParticipantFactory::get_instance()->load_XML_profiles_file(xml_path);
		if (ReturnCode_t::RETCODE_OK != ret)
		{
			return false;
		}
		participant_ =
			DomainParticipantFactory::get_instance()->create_participant_with_profile(12, participant_profile);
		if (!participant_)
		{
			return false;
		}
		type_.register_type(participant_);
	}
	Topic *topic = nullptr;
	Subscriber *subscriber = nullptr;
	DataReader *datareader = nullptr;
	do
	{
		topic = participant_->create_topic_with_profile(topic_name, "Message", topic_profile);
		if (!topic)
		{
			break;
		}

		subscriber = participant_->create_subscriber(SUBSCRIBER_QOS_DEFAULT);
		if (!subscriber)
		{
			break;
		}

		auto listener = new SubListener();
		listener->callback = callbk;
		datareader = subscriber->create_datareader_with_profile(topic, datareader_profile, listener);
		if (!datareader)
		{
			break;
		}
		auto qos = datareader->get_qos().reliability();
		subscribers_[topic_name] = subscriber;
		return true;
	} while (false);

	if (topic != nullptr)
	{
		participant_->delete_topic(topic);
	}
	if (datareader != nullptr)
	{
		auto listener = subscriber->get_listener();
		delete listener;
		subscriber->delete_datareader(datareader);
	}
	if (subscriber != nullptr)
	{
		participant_->delete_subscriber(subscriber);
	}

	return false;
}

void Subscriber_profile::run(std::string logFile)
{
}

void Subscriber_profile::stop()
{
}
void Subscriber_profile::SubListener::on_data_available(
	eprosima::fastdds::dds::DataReader *reader)
{
	SampleInfo info;
	while (reader->take_next_sample(&message_, &info) == ReturnCode_t::RETCODE_OK)
	{
		if (info.instance_state == ALIVE_INSTANCE_STATE)
		{
			Logger::getInstance().info(std::to_string(message_.index()) + " " + std::to_string(info.reception_timestamp.to_ns()) + " " + std::to_string(info.source_timestamp.to_ns())
			+ " " + std::to_string(message_.timestamp()) + " " + std::to_string(info.reception_timestamp.to_ns() - info.source_timestamp.to_ns()) + " " + reader->get_topicdescription()->get_name());
			this->callback(message_);
		}
	}
}
void Subscriber_profile::SubListener::on_subscription_matched(
	eprosima::fastdds::dds::DataReader *reader,
	const eprosima::fastdds::dds::SubscriptionMatchedStatus &info)
{

	std::string topic_name = reader->get_topicdescription()->get_name();
#if 1
	runLog(topic_name);
#endif
	//std::cout << "subscription matched:" << topic_name << std::endl;
}

LogMsg::~LogMsg()
{
}

void LogMsg::init(std::string logFile)
{
	ofs.open(logFile, std::ios_base::out);
	running_.store(true);
}

void LogMsg::appendMsg(const Message &msg)
{
	time_t curTime = getTimestamp();
	std::unique_lock<std::mutex> locker(mutex_);
	strMsg.append(std::to_string(msg.index()) + "," + std::to_string(msg.timestamp()) + "," + std::to_string(curTime) + "\n");
	if (strMsg.size() >= 1024 * 2)
	{
		cond_.notify_all();
	}
}

void LogMsg::run()
{
	while (running_.load())
	{
		std::unique_lock<std::mutex> locker(mutex_);
		cond_.wait(locker);
		ofs << strMsg;
		// ofs.flush();
		strMsg.clear();
	}
}

void LogMsg::stop()
{
	running_.store(false);
	cond_.notify_all();
}